package io.renren.modules.medical.service;

import com.alibaba.fastjson.JSONObject;
import io.renren.common.exception.MedicalException;
import io.renren.common.result.ResultCodeEnum;
import io.renren.modules.medical.entity.WebSocketInterEntity;
import io.renren.modules.medical.service.impl.WebSocketInterServiceImpl;
import lombok.extern.slf4j.Slf4j;
import javax.annotation.PostConstruct;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Slf4j
@ServerEndpoint(value = "/medical/webSocket/{sid}")
@Component
// TODO 使用redis优化
public class WebSocketServer {

    @Value("${threadPool.corePoolSize}")
    private Integer CORE_POOL_SIZE;

    @Value("${threadPool.maximumPoolSize}")
    private Integer MAXIMUM_POOL_SIZE;

    @Value("${threadPool.keepAliveTime}")
    private Long KEEP_ALIVE_TIME;

    @Value("${threadPool.capacity}")
    private Integer CAPACITY;
    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    //接收用户的sid，指定需要推送的用户
    private String sid;
    //新：使用map对象优化，便于根据sid来获取对应的WebSocket
    private static final ConcurrentHashMap<String, WebSocketServer> websocketMap = new ConcurrentHashMap<>();

    private static ThreadPoolExecutor sendExecutor;
    private static ThreadPoolExecutor storageExecutor;


    // 创建组件时调用的方法
    @PostConstruct
    public void init() {

        log.info("WebSocketServer初始化...");
        sendExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAXIMUM_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(CAPACITY),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        storageExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAXIMUM_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(CAPACITY),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }


    /**
     * 连接成功后调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        this.session = session;
        if (StringUtils.isEmpty(sid)) {
            throw new MedicalException(ResultCodeEnum.SID_ERROR.getCode(), "sid为null");
        }

        websocketMap.put(sid, this); //加入map中
        this.sid = sid;               //给此线程的sid赋值
        log.info("有新窗口开始监听,sid：" + sid + ",当前在线人数为：" + getOnlineCount());
        try {
            sendMessage("$pong$"); // 测试是否成功连接服务器和客户端
        } catch (IOException e) {
            // 抛出异常避免服务器宕机
            log.error("websocket IO异常");
            e.printStackTrace();
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        System.out.println(sid);
        if (this.sid != null && websocketMap.get(this.sid) != null) {
            websocketMap.remove(this.sid);  //从map中删除
            // 将map中的数据全部保存到数据库中
            log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
        }
    }

    /**
     * 收到客户端消息后调用的方法，根据业务要求进行处理，这里就简单地将收到的消息直接群发推送出去
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message) {
        // 对正常信息进行处理
        if (StringUtils.isEmpty(message)) {
            return;
        }
        if (message.equals("$ping$")) {
            // 心跳设置，判断连接是否正常进行
            try {
                String res = "$pong$";
                sendInfo(res, this.sid);    // 假定心跳返回的字符串为1$
                return;
            } catch (IOException e) {
                log.error("心跳消息发送失败");
                e.printStackTrace();
            }

        }
        Map<String, Object> resMap = JSONObject.parseObject(message);
        HashMap<String, Object> errorData = new HashMap<>();
        HashMap<String, Object> respData = new HashMap<>();
        // 对正常信息进行处理
        try {
            //从message中解析出toSid和content
            String toSid = String.valueOf(resMap.get("sendToId"));
            String content = String.valueOf(resMap.get("content"));
            if (StringUtils.isEmpty(toSid) || StringUtils.isEmpty(content)) {
                return;
            }
            // 创建消息对象
            WebSocketInterEntity webSocketInterEntity = new WebSocketInterEntity();
            webSocketInterEntity.setSendId(Long.valueOf(this.sid));
            webSocketInterEntity.setSendToId(StringUtils.isNotEmpty(toSid) ? Long.parseLong(toSid) : 0L);
            webSocketInterEntity.setMessage(content);
            webSocketInterEntity.setSendTime(new Date());

            if (websocketMap.get(toSid) == null) {
                log.info("对方未上线");
                // 未上线的处理，未上线直接保存信息到数据库
                webSocketInterEntity.setIsVisited(0);
            } else {
                // 当前用户在线
                webSocketInterEntity.setIsVisited(1);
                sendExecutor.submit(() -> {
                    // 发送消息到对方用户
                    try{
                        log.info("进入读线程，当前线程的id为：{}，要写入的信息是{}", Thread.currentThread().getId(), content);
                        respData.put("data", content);
                        respData.put("sendId", this.sid);
                        respData.put("code", ResultCodeEnum.SUCCESS.getCode());
                        sendInfo(JSONObject.toJSONString(respData), toSid);
                        return "消息发送成功";
                    }catch (Exception e){
                        errorData.put("code", ResultCodeEnum.MESSAGE_SAVE_ERROR.getCode());
                        errorData.put("msg", "消息发送失败");
                        sendInfo(JSONObject.toJSONString(errorData), sid);
                        throw new MedicalException(ResultCodeEnum.MESSAGE_SAVE_ERROR.getCode(), "消息发送失败");
                    }
                });
            }

            storageExecutor.submit(() -> {
                // 保存消息到数据库
                try {
                    // 保存消息到数据库
                    log.info("进入写线程，当前线程的id为：{}，要写入的信息是{}", Thread.currentThread().getId(), webSocketInterEntity);
                    WebSocketInterService webSocketInterService = new WebSocketInterServiceImpl();
                    List<WebSocketInterEntity> list = new ArrayList<>();
                    list.add(webSocketInterEntity);
                    webSocketInterService.saveBatch(list);
                    return "消息保存成功";
                } catch (Exception e) {
                    // 捕获异常并将其转换为运行时异常
                    errorData.put("code", ResultCodeEnum.MESSAGE_SAVE_ERROR.getCode());
                    errorData.put("msg", "消息保存失败");
                    sendInfo(JSONObject.toJSONString(errorData), sid);
                    throw new MedicalException(ResultCodeEnum.MESSAGE_SAVE_ERROR.getCode(), "消息保存失败");
                }
            });
        } catch (Exception e) {
            log.error("onMessage消息发送出现异常");
            e.printStackTrace();
        }
    }

    /**
     * 发生错误时的回调函数
     */
    @OnError
    public void onError(Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }

    /**
     * 实现服务器主动推送消息
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 发送消息（用set会方便些）
     */
    public static void sendInfo(String message, String sid) throws IOException {
        if (StringUtils.isEmpty(message)) {
            return;
        }
        try {
            if (websocketMap.get(sid) != null && websocketMap.get(sid).session.isOpen()) {
                WebSocketServer server = websocketMap.get(sid);
                server.sendMessage(message);
            }
        } catch (IOException e) {
            log.error("sendInfo消息发送出现异常");
            e.printStackTrace();
        }

    }

    public static synchronized int getOnlineCount() {
        return websocketMap.keySet().size();
    }
}
